package cn.hy.threads.transaction;

import cn.hutool.core.util.IdUtil;
import jdk.vm.ci.code.site.Call;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 原理
 * 使用了CountDownLatch + 编程式事务 架构。
 * <p>
 * 实例化两个CountDownLatch 对象，一个标明子线程运行完成，另一个标明下一步操作。
 * 当子线程完成增删改后，调用await方法由主线程控制下一步操作。主线程调用countDown，
 * 激活子线程，子线程判断变量isCommit 是否为真，真则提交，假则回滚。
 *
 * <p>
 * 就是模仿2PC 设计模式开发。
 *
 * 批量操作同一操作
 * 1.多线程入库仅支持同一种操作:比如 单insert/ 单update ,由于数据库的事务隔离级别,不能对同一条数据同时insert和update,会导致死锁。
 * 2.如果非要使用多线程同时insert 和 update,则需要把事务隔离级别降为READ_COMMITTED,但是会出现先update导致失败问题(注解不存在)
 * 3.MySQL默认事务隔离级别:REPEATABLE_READ
 *
 * 基于 spring 事务管理器
 *
 */
@Component
public class RunnableTransactionManager {

    private final Logger logger = LoggerFactory.getLogger(RunnableTransactionManager.class);

    // DataSourceTransactionManager
    // 可以直接拿来用
    @Resource
    private PlatformTransactionManager transactionManager;

    @Resource
    private ThreadPoolExecutor executor;


    private final AtomicBoolean isCommit = new AtomicBoolean(true);


    public void executeAllTasks(List<Task> tasks) {

        List<Runnable> runList = new ArrayList<>();

        final CountDownLatch taskCountLatch = new CountDownLatch(tasks.size());

        for (Task task : tasks) {

            runList.add(new Runnable() {

                @Override
                public void run() {
                    String messageId = IdUtil.fastSimpleUUID();

                    DefaultTransactionDefinition def = new DefaultTransactionDefinition();
                    // 开启 新事务，挂起当前事务
                    def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
                    TransactionStatus status = transactionManager.getTransaction(def);
                    try {
                        task.run();
                        logger.info("业务执行成功,业务id:{}", messageId);
                    } catch (Exception e) {
                        logger.error("执行任务:{} 失败", messageId, e);
                        isCommit.compareAndSet(true, false);
                    } finally {
                        // 保证都可以减1
                        taskCountLatch.countDown();

                        if (isCommit.get()) {
                            transactionManager.commit(status);
                        } else {
                            transactionManager.rollback(status);
                        }
                    }
                }
            });
        }

        try {

            for (Runnable runnable : runList) {
                executor.execute(runnable);
            }
            taskCountLatch.await();
        } catch (Exception e) {
            isCommit.compareAndSet(true, false);
            logger.error(" 统一提交失败  ", e);
        }
    }

}